#!/home/aaron/pyenv/bin/python
from supply_test import SupplyTask
import unittest
import utils
from datetime import datetime, date,timedelta
import os
# 编写子类继承unittest.TestCase
from subprocess import CalledProcessError
from pathlib import Path
import time


class TestSupplyProject(unittest.TestCase):

    ##测试专用
    st = SupplyTask(task_no=2631, biz_dt=date.today(), isLast=False, test_mode=True)

    def test_get_local_ip(self):
        self.assertEqual(utils.get_local_ip(), "192.168.162.21")

    def test_replace_yyyymmdd(self):
        text = "path/#yyyymmdd#"
        ddate = date(2020, 1, 21)
        new = utils.replace_yyyymmdd(text, ddate)
        self.assertEqual(new, "path/20200121")

        text = "path/#MMdd#"
        ddate = date(2020, 1, 21)
        new = utils.replace_yyyymmdd(text, ddate)
        self.assertEqual(new, "path/0121")

        text = "path/#yyyyMMdd+1d#"
        ddate = date(2020, 1, 31)
        new = utils.replace_yyyymmdd(text, ddate)
        self.assertEqual(new, "path/20200201")

        text = "path/#yyyyMMdd-2D#"
        ddate = date(2020, 2, 1)
        new = utils.replace_yyyymmdd(text, ddate)
        self.assertEqual(new, "path/20200130")

        text = "path/#yyyyMMdd-2D##TIME%s#"
        ddate = date(2020, 2, 1)
        new = utils.replace_yyyymmdd(text, ddate)
        self.assertEqual(len(new), len("path/20200130") + 10)

    def test_init(self):
        self.st.vld_flg = 1
        self.st.cfg_vld_flg = 1
        self.st.dtbs_tp = "postgresql"
        self.assertEqual(self.st.dtbs_tp, "postgresql")

    def test_need_execute(self):
        self.st.vld_flg = 0
        self.assertFalse(self.st.need_execute())
        self.st.vld_flg = 1
        self.assertTrue(self.st.need_execute())


    def test_export_last(self):
        self.st.isLast = True
        self.st.touch_finish_flag()
        self.assertTrue(self.st.data_file.with_name("finish.flag").exists())

    def notest_export_diff(self):
        '''
         test write_file_all_str
        '''

        st1 = SupplyTask(task_no=914, biz_dt=date.today() - timedelta(days=1), isLast=False, test_mode=True)
        origin_file = st1.data_file
        st1.local_file_abs_path = st1.local_file_abs_path + ".test"
        st1.data_file = Path(st1.local_file_abs_path)
        st1.export()
        one = origin_file.stat()
        two = st1.data_file.stat()
        self.assertEqual(one.st_size, two.st_size)
        self.assertEqual(one.st_mode, two.st_mode)
        self.assertEqual(one.st_dev, two.st_dev)
        self.assertEqual(one.st_uid, two.st_uid)
        self.assertEqual(one.st_gid, two.st_gid)

    def notest_export_diff2(self):
        '''
         test copy_to_file
        '''

        st1 = SupplyTask(task_no=2575, biz_dt=date.today() - timedelta(days=1), isLast=False, test_mode=True)
        origin_file = st1.data_file
        st1.local_file_abs_path = st1.local_file_abs_path + ".test"
        st1.data_file = Path(st1.local_file_abs_path)
        st1.export()
        one = origin_file.stat()
        two = st1.data_file.stat()
        self.assertEqual(one.st_size, two.st_size)
        self.assertEqual(one.st_mode, two.st_mode)
        self.assertEqual(one.st_dev, two.st_dev)
        self.assertEqual(one.st_uid, two.st_uid)
        self.assertEqual(one.st_gid, two.st_gid)

    def notest_export_diff3(self):
        '''
         test write_to_file
        '''
        st1 = SupplyTask(task_no=1304, biz_dt=date.today() - timedelta(days=1), isLast=False, test_mode=True)
        origin_file = st1.data_file
        st1.local_file_abs_path = st1.local_file_abs_path + ".test"
        st1.data_file = Path(st1.local_file_abs_path)
        st1.export()
        one = origin_file.stat()
        two = st1.data_file.stat()
        self.assertEqual(one.st_size, two.st_size)
        self.assertEqual(one.st_mode, two.st_mode)
        self.assertEqual(one.st_dev, two.st_dev)
        self.assertEqual(one.st_uid, two.st_uid)
        self.assertEqual(one.st_gid, two.st_gid)

    def test_export_copy_to_file_empty(self):
        st1 = SupplyTask(task_no=2631, biz_dt=date.today() - timedelta(days=1), isLast=False, test_mode=True)
        st1.sql_text = "select 1 from dw.f_cm_branch where 1 = 2"
        if st1.data_file.exists():
            st1.data_file.unlink()
        st1.export()
        self.assertTrue(st1.data_file.exists())


    def test_transfer_scp(self):
        st1 = SupplyTask(task_no=2631, biz_dt=date.today() - timedelta(days=1), isLast=False, test_mode=True)
        st1.prvd_data_mod = 'scp'
        st1.transfer_cli = st1.get_transfer_cli(st1.prvd_data_mod, st1.ip_addr,st1.usr_nm,st1.psw,st1.cd_mod)
        st1.upl_flg = 1
        st1.sql_text = "select * from dw.f_cm_branch where 1 = 1"
        st1.export()
        st1.upload(st1.data_file.as_posix(),st1.remote_data_file.as_posix())
        test_file = st1.data_file.with_name(st1.file_nm+".test")
        st1.transfer_cli.get(test_file.as_posix(), st1.remote_data_file.as_posix()) 
        self.assertEqual(st1.data_file.stat().st_size, test_file.stat().st_size)

    def test_transfer_sftp(self):
        time.sleep(5)
        st1 = SupplyTask(task_no=2631, biz_dt=date.today() - timedelta(days=1), isLast=False, test_mode=True)
        st1.prvd_data_mod = 'sftp'
        st1.transfer_cli = st1.get_transfer_cli(st1.prvd_data_mod, st1.ip_addr,st1.usr_nm,st1.psw,st1.cd_mod)
        st1.upl_flg = 1
        st1.sql_text = "select * from dw.f_cm_branch where 1 = 1"
        st1.data_file.write_text("testtest测试数据撤销\n测试去玩儿安抚阿斯蒂芬就就了阿斯蒂芬阿萨德.")
        st1.upload(st1.data_file.as_posix(),st1.remote_data_file.as_posix())
        test_file = st1.data_file.with_name("test_sftp.txt.test")
        st1.transfer_cli.get(test_file.as_posix(), st1.remote_data_file.as_posix()) 
        self.assertEqual(st1.data_file.stat().st_size, test_file.stat().st_size)

    def test_transfer_ftp(self):
        st1 = SupplyTask(task_no=2631, biz_dt=date.today() - timedelta(days=1), isLast=False, test_mode=True)
        st1.prvd_data_mod = 'ftp'
        st1.usr_nm = 'dwtrans'
        st1.psw = 'dwtrans'
        st1.cd_mod = 'gbk'
        st1.data_file = st1.data_file.with_name("测试test_ftp.txt")
        st1.remote_data_file = Path("/home/file/output/testftp/11/22/33/20210608") / "测试test_ftp.txt"
        st1.transfer_cli = st1.get_transfer_cli(st1.prvd_data_mod, st1.ip_addr,st1.usr_nm,st1.psw,st1.cd_mod)
        st1.upl_flg = 1
        st1.sql_text = "select * from dw.f_cm_branch where 1 = 1"
        st1.data_file.write_text("testtest测试数据撤销\n测试去玩儿安抚阿斯蒂芬就就了阿斯蒂芬阿萨德.")
        st1.upload(st1.data_file.as_posix(),st1.remote_data_file.as_posix())
        test_file = st1.data_file.with_name("测试test_ftp.txt.test")
        st1.transfer_cli.get(test_file.as_posix(), st1.remote_data_file.as_posix()) 
        self.assertEqual(st1.data_file.stat().st_size, test_file.stat().st_size)

  

if __name__ == "__main__":
    unittest.main()
